package com.heima.kafka.boot.stream;

import com.heima.kafka.boot.config.KafkaStreamListener;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Component
public class StreamListener implements KafkaStreamListener<KStream<String,String>> {
    @Override
    public String listenerTopic() {
        return "itcast-topic-input";
    }

    @Override
    public String sendTopic() {
        return "itcast-topic-output";
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return stream.flatMapValues( value -> {
            // value 接收消息的具体内容
            return Arrays.asList(value.split(" "));
        })
                // 根据value进行分组
                .groupBy((key,value)->value)
                // 聚合计算时间间隔
                .windowedBy(TimeWindows.of(5000))
                // 聚合查询：求单词总个数
                .count()
                // 转成 KStream
                .toStream()
                // key和value转成string
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), value.toString());
                });
    }
}
